package com.example.core.debezium.listener;

import com.example.core.cluster.MasterServerService;
import com.example.core.debezium.message.ChangeData;
import com.example.core.debezium.message.Message;
import com.example.core.utils.JSONUtils;
import io.debezium.config.Configuration;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.Json;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

@Service
public class SysDebeziumListenerImpl implements SysDebeziumListener, MasterServerService {

    Logger logger = LoggerFactory.getLogger(getClass());

    @Value("${spring.datasource.server.name}")
    private String servername;

    @Value("${spring.datasource.username}")
    private String username;

    @Value("${spring.datasource.password}")
    private String password;

    @Value("${spring.datasource.db}")
    private String db;

    @Value("${spring.datasource.host}")
    private String host;

    @Value("${spring.datasource.port}")
    private int port;

    @Value("${spring.datasource.serverTimezone}")
    private String serverTimezone;

    @Value("${server.port}")
    private int serverId;

    private DebeziumEngine engine;

    @Resource
    private List<SysDebeziumDataChangeListener> sysDebeziumDataChangeListeners = new ArrayList<>();

    @Override
    public void receiveChangeEvent(String value) {
        if (Objects.nonNull(value)) {
            Map<String, Object> payload = JSONUtils.getPayload(value);
            String handleType = JSONUtils.getHandleType(payload);
            if (!("NONE".equals(handleType) || "READ".equals(handleType))) {
                ChangeData changeData = JSONUtils.getChangeData(payload);
                Map<String, Object> data;
                if ("DELETE".equals(handleType)) {
                    data = changeData.getBefore();
                } else {
                    data = changeData.getAfter();
                }
                Message message = new Message();
                message.setData(data);
                message.setDbType("MYSQL");
                message.setDatabase(String.valueOf(changeData.getSource().get("db")));
                message.setTable(String.valueOf(changeData.getSource().get("table")));
                message.setHandleType(handleType);
                for (SysDebeziumDataChangeListener sysDebeziumDataChangeListener : sysDebeziumDataChangeListeners) {
                    sysDebeziumDataChangeListener.changeEvent(message);
                }
            }
        }
    }

    @Override
    public void start() {
        Executor executor = Executors.newSingleThreadExecutor();
        Configuration config = Configuration.create()
                .with("connector.class", "io.debezium.connector.mysql.MySqlConnector")
                .with("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore")
                .with("offset.storage.file.filename", "/tmp/offset.dat")
                .with("offset.flush.interval.ms", 60000)
                .with("name", "my-sql-connector")
                .with("database.hostname", host)
                .with("database.port", port)
                .with("database.user", username)
                .with("database.password", password)
                .with("database.connectionTimeZone", serverTimezone)
                .with("database.server.id", serverId)
                .with("database.include.list", db)
                .with("snapshot.mode", "initial")
                .with("database.server.name", servername)
                .with("database.history", "io.debezium.relational.history.FileDatabaseHistory")
                .with("database.history.file.filename", "/tmp/history.dat")
                .build();

        this.engine = DebeziumEngine.create(Json.class)
                .using(config.asProperties())
                .notifying(record -> {
                    receiveChangeEvent(record.value());
                }).build();

        executor.execute(engine);
    }

    @Override
    public void stop() {

    }
}
